package mr.stocks.question3;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

import mr.stocks.avro.model.StockPriceByYearReportAvro;
import mr.stocks.avro.model.StockPriceByYearValueAvro;

import org.apache.avro.mapred.AvroValue;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class StockPriceByYearAvroCombiner extends
        Reducer<Text, AvroValue<StockPriceByYearReportAvro>, 
        		Text, AvroValue<StockPriceByYearReportAvro>> {
 
	private AvroValue<StockPriceByYearReportAvro> outValueAvro = new AvroValue<StockPriceByYearReportAvro>();
	private StockPriceByYearReportAvro outValue = new StockPriceByYearReportAvro();
	private Map<CharSequence, StockPriceByYearValueAvro> priceMap = new HashMap<CharSequence, StockPriceByYearValueAvro>();

    @Override
    protected void reduce(Text inKey, Iterable<AvroValue<StockPriceByYearReportAvro>> inValues, Context context)
            throws IOException, InterruptedException {

    	priceMap.clear();

    	for(AvroValue<StockPriceByYearReportAvro> value : inValues){
    		
    		Map<CharSequence, StockPriceByYearValueAvro> inMap = value.datum().getAverageAnnualPrice();
    		
    		for(CharSequence key: inMap.keySet()){
    			if(priceMap.containsKey(key))
    			{
    				StockPriceByYearValueAvro stockAnnaulValue = 
    						priceMap.get(key);
    				
    				stockAnnaulValue.setCount(stockAnnaulValue.getCount() + inMap.get(key).getCount());
    				stockAnnaulValue.setPrice(stockAnnaulValue.getPrice() + inMap.get(key).getPrice());
    			}
    			else
    			{
    				priceMap.put(key, inMap.get(key));
    			}
    		}
    	}
    	
    	outValue.setStockSymbol(inKey.toString());
    	outValue.setAverageAnnualPrice(priceMap);
    	outValueAvro.datum(outValue);
    	context.write(inKey, outValueAvro);
    }
}
